package com.shujia.jinjie;

import com.shujia.utils.HBaseUtil;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.ValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import redis.clients.jedis.Jedis;

import java.util.Scanner;
import java.util.Set;

/*
    使用redis第三方的存储工具存储hbase索引（本质依旧是列值与行键产生映射关系）
 */
public class HBaseWithRedisIndex {
    //1、获取hbase数据库连接对象和操作对象
    static Connection conn = HBaseUtil.CONNECTION;
    static Admin admin = HBaseUtil.ADMIN;

    //获取redis连接对象
    static Jedis jedis = new Jedis("192.168.220.100", 12346);

    public static void main(String[] args) throws Exception {
        //步骤1：在redis中构建映射关系（性别：学号）
//        buildIndexInRedis();

        //使用：先通过查询redis中性别对应的学号，拿着学号去hbase原表中查询获取结果
        Scanner sc = new Scanner(System.in);
        System.out.println("请输入您要查询的性别：");
        String gender = sc.next();
        selectGenderFromHbase(gender);
    }

    public static void selectGenderFromHbase(String gender) throws Exception {
        if ("男".equals(gender)) {
            selectIdFromRedis(gender);
        } else if ("女".equals(gender)) {
            selectIdFromRedis(gender);
        } else {
            System.out.println("没有该性别");
        }
    }

    //单独编写一个方法查询redis
    public static void selectIdFromRedis(String gender) throws Exception {
        Table students2 = conn.getTable(TableName.valueOf("students2"));

        Set<String> ids = jedis.smembers("性别:"+gender);
        for (String id : ids) {
            Result result = students2.get(new Get(Bytes.toBytes(id)).addColumn(Bytes.toBytes("info"), Bytes.toBytes("name")));
            String name = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name")));
            System.out.println("学号:" + id + "，姓名:" + name);
        }
    }

    public static void buildIndexInRedis() throws Exception {

        //获取要构建索引的原表
        Table students2 = conn.getTable(TableName.valueOf("students2"));

        Scan scan = new Scan();
        //获取男生的学号，放入到redis中
        //创建列值过滤器
        ValueFilter filter1 = new ValueFilter(CompareOperator.EQUAL, new BinaryComparator(Bytes.toBytes("男")));
        scan.setFilter(filter1);
        ResultScanner resultScanner = students2.getScanner(scan);
        for (Result result : resultScanner) {
            //获取每一行的行键即可
            String id = Bytes.toString(result.getRow());
            //将学号以值的方式添加到redis键对应的值中
            //因为男生的学号有很多个，且不重复，所以我们在redis中采用set的数据类型存储
            jedis.sadd("性别:男", id);
        }

        //获取男生的学号，放入到redis中
        //创建列值过滤器
        ValueFilter filter2 = new ValueFilter(CompareOperator.EQUAL, new BinaryComparator(Bytes.toBytes("女")));
        scan.setFilter(filter2);
        ResultScanner resultScanner2 = students2.getScanner(scan);
        for (Result result : resultScanner2) {
            //获取每一行的行键即可
            String id = Bytes.toString(result.getRow());
            //将学号以值的方式添加到redis键对应的值中
            //因为男生的学号有很多个，且不重复，所以我们在redis中采用set的数据类型存储
            jedis.sadd("性别:女", id);
        }


    }
}
